package com.wuwangfu.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author jcshen
 * @Date 2023-03-01
 * @PackageName:com.wuwangfu.state
 * @ClassName: KeyedStated
 * @Description: 设置重启策略
 * @Version 1.0.0
 *
 * 1、State
 *      flink实时计算程序为了保证计算过程，出现异常可以容错，就要将中间的计算结果数据存储起来，这些中间结果数据就叫状态。
 *      State可以是多种类型，默认保存在JobManager的内存中，也可以保存到TaskManager本地文件系统或HDFS分布式文件系统。
 * 2、StateBackEnd
 *      用来保存state的存储后端就叫StateBackEnd，默认保存在JobManager的内存中，也可以保存在本地文件系统或HDFS分布式文件系统。
 * 3、Checkpoint
 *      Flink实时计算为了容错，可以将中间数据定期保存起来，这种定期触发保存中间结果的机制叫Checkpoint。Checkpoint是周期执行的，
 *      具体的过程是JobManager定期向TaskManager中的SubTask发送RPC消息，SubTask将其计算的State保存到StateBackEnd中，并且向
 *      JobManager响应Checkpoint是否成功。如果程序出现异常或重启，TaskManager中的SubTask可以从上一次成功的Checkpoint的State中恢复。
 * 4、重启策略
 *      Flink实时计算程序，为了容错，需要开启Checkpoint，一旦开启Checkpoint，如果没有重启策略，默认的重启策略是无限重启，可以设置其它重启策略。
 *      如：重启固定次数且可以延迟执行的策略。
 * 5、CheckpointingMode
 *
 *
 */
public class RestartStrategy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置重启策略：重启3次，每次延迟5秒
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000));
        //故障率，不超过3次，30秒以内，延迟3秒
        //env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.seconds(30),Time.seconds(3)));

        /*开启checkpoint，默认无限重启策略*/
        env.enableCheckpointing(10000);//10秒一次

        /**
         * hadoop,1
         * hadoop,1
         * error,1
         */
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> flated = line.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] fields = value.split(",");
                if ("error".equals(fields[0])){
                    throw new RuntimeException("出现异常");
                }
                out.collect(Tuple2.of(fields[0], Integer.parseInt(fields[1])));
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyed = flated.keyBy(t -> t.f0);
        keyed.sum(1).print();

        env.execute();
    }
}
